package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.FunnelAnalysisBean;
import com.atguigu.edu.realtime.common.kafkaTopics;
import com.atguigu.edu.realtime.util.KafkaUtil;
import org.apache.commons.collections.MapUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.client.program.StreamContextEnvironment;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 漏洞分析
 */
public class DwsTrafficFunnelAnalysisWindow {
    public static void main(String[] args) throws Exception {


        // TODO: 2022/9/3 获取页面日志 根据 pageId = home 或者 course_detail 构建bean
        StreamExecutionEnvironment env = StreamContextEnvironment.getExecutionEnvironment();

        String topic_name = kafkaTopics.DWD_TRAFFIC_PAGE_LOG;
        String groupId = "dws_traffic_funnel_analysis_window";
        /**
         * {
         *     "common":{
         *         "sc":"1",
         *         "ar":"12",
         *         "uid":"75",
         *         "os":"iOS 13.2.9",
         *         "ch":"Appstore",
         *         "is_new":"1",
         *         "md":"iPhone 8",
         *         "mid":"mid_324",
         *         "vc":"v2.1.134",
         *         "ba":"iPhone",
         *         "sid":"1c052f29-79b2-4665-85ec-8fc902a39f62"
         *     },
         *     "page":{
         *         "page_id":"exam",
         *         "item":"210",
         *         "during_time":1351000,
         *         "item_type":"paper_id",
         *         "last_page_id":"course_detail"
         *     },
         *     "ts":1662200137091
         * }
         */
        SingleOutputStreamOperator<FunnelAnalysisBean> logStream = env.addSource(KafkaUtil.getKafkaConsumer(topic_name, groupId))
                .flatMap(new FlatMapFunction<String, FunnelAnalysisBean>() {
                    @Override
                    public void flatMap(String value, Collector<FunnelAnalysisBean> out) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        JSONObject page = jsonObject.getJSONObject("page");
                        long ts = MapUtils.getLongValue(jsonObject, "ts");
                        String page_id = MapUtils.getString(page, "page_id", "");

                        //首页浏览
                        if ("home".equals(page_id)) {
                            FunnelAnalysisBean analysisBean = FunnelAnalysisBean.builder()
                                    .homePageUserCount(1L)
                                    .ts(ts) //毫秒值
                                    .build();

                            out.collect(analysisBean);

                        }
                        // 订单详情页
                        else if ("course_detail".equals(page_id)) {
                            FunnelAnalysisBean analysisBean = FunnelAnalysisBean.builder()
                                    .itemDetailPageUserCount(1L)
                                    .ts(ts) //毫秒值
                                    .build();
                            out.collect(analysisBean);
                        }
                    }
                });

        // TODO: 2022/9/3 对接加购事实表 转化bean

        // TODO: 2022/9/3 对接下单事实表 转化bean

        // TODO: 2022/9/3 对接支付成功事实表  转化bean


        // TODO: 2022/9/3 上述四个流union

        // TODO: 2022/9/3 设置watermark 再开窗  将结果写入clickhouse

        env.execute();
    }
}
